package com.atguigu.flink.chapter06;

import com.atguigu.flink.bean.AdsClickLog;
import com.atguigu.flink.bean.OrderEvent;
import com.atguigu.flink.bean.TxEvent;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/10 16:24
 */
public class Flink09_OrderTx {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 1.读取数据:准备两条流
        SingleOutputStreamOperator<OrderEvent> orderDS = env
                .readTextFile("F:\\atguigu\\01_course\\code\\flink210323\\input\\OrderLog.csv")
                .map(r -> {
                    String[] datas = r.split(",");
                    return new OrderEvent(
                            Long.parseLong(datas[0]),
                            datas[1],
                            datas[2],
                            Long.parseLong(datas[3])
                    );
                });

        SingleOutputStreamOperator<TxEvent> txDS = env
                .readTextFile("F:\\atguigu\\01_course\\code\\flink210323\\input\\ReceiptLog.csv")
                .map(r -> {
                    String[] datas = r.split(",");
                    return new TxEvent(
                            datas[0],
                            datas[1],
                            Long.parseLong(datas[2])
                    );
                });

        // 处理数据
        // 1.通过 txid 关联两条流 ： connect
        ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderDS.connect(txDS);

        // TODO 一般使用connect关联的时候，要对 关联条件 keyby，避免 走错房间
        // 也可以两条流，自己分别做 keyby，都得到 keyedstream， 之后再 connect，也可以
        ConnectedStreams<OrderEvent, TxEvent> orderTxKCS = orderTxCS.keyBy(order -> order.getTxId(), tx -> tx.getTxId());


        // 2.考虑 process 实现
        SingleOutputStreamOperator<String> resultDS = orderTxKCS.process(new CoProcessFunction<OrderEvent, TxEvent, String>() {

            // 用来存储 交易数据，key = 交易码， value = 交易数据
            Map<String, TxEvent> txMap = new HashMap<String, TxEvent>();
            // 用来存储 业务数据，key = 交易码， value = 业务数据
            Map<String, OrderEvent> orderMap = new HashMap<String, OrderEvent>();

            /**
             * 处理 业务数据 的逻辑
             * @param value
             * @param ctx
             * @param out
             * @throws Exception
             */
            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                // 当前来的是 业务数据， 查看一下 交易数据 来过没有
                if (txMap.containsKey(value.getTxId())) {
                    // 1. 交易数据 来过 ===》 对账成功
                    out.collect("订单：" + value.getOrderId() + " 对账成功！");
                    // 清理 存在 Map的 交易数据
                    txMap.remove(value.getTxId());
                } else {
                    // 2. 交易数据 没来过 ===> 把自己 存起来， 等待 交易数据
                    orderMap.put(value.getTxId(), value);
                }
            }

            /**
             * 处理 交易数据 的逻辑
             * @param value
             * @param ctx
             * @param out
             * @throws Exception
             */
            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                // 当前来的是 交易数据， 查看一下 业务数据 来过没有
                if (orderMap.containsKey(value.getTxId())) {
                    // 1. 业务数据 来过 ===》 对账成功
                    out.collect("订单：" + orderMap.get(value.getTxId()) + " 对账成功！");
                    // 清理 存在 Map的 业务数据
                    orderMap.remove(value.getTxId());
                } else {
                    // 2. 业务数据 没来过 ===> 把自己 存起来， 等待 业务数据
                    txMap.put(value.getTxId(), value);
                }
            }
        });

        resultDS.print();


        env.execute();
    }
}
